#include "XTcpServer.h"
#include "XTcpTask.h"
#include "XThreadPool.h"
#include <iostream>

#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <event2/thread.h>

using namespace std;

XTcpServer::XTcpServer(int threadCount, std::string ip, int port)
{
    m_args.threadCount = threadCount;
    m_args.ip = ip;
    m_args.port = port;

    /**
     * 多线程环境下没有调用evthread_use_windows_threads或evthread_use_pthreads函数
     * 会导致event_base_dispatch函数一直阻塞，(本demo分为主线程和dispatch线程)
     * 即使调用了event_base_loopbreak或event_base_loopexit
     * 也无法让event_base_dispatch退出事件循环。
     */
    evthread_use_pthreads();
}

bool XTcpServer::construct()
{
    bool ret = true;
    if(!m_args.isRunning)
    {
        ret = ret && XThreadPool::GetInstance()->Init(m_args.threadCount);

        /* 创建libevent的上下文 */
        m_args.base = event_base_new();
        if (!m_args.base)
        {
            cout << "event_base_new  error!" << endl;
        }

        //socket ，bind，listen 绑定事件
        sockaddr_in sin = {0};
        sin.sin_family = AF_INET;                            //设置地址家族
        sin.sin_port = htons(m_args.port);                   //设置端口
        sin.sin_addr.s_addr = inet_addr(m_args.ip.c_str());  //设置地址

        m_args.evConnListener = evconnlistener_new_bind(m_args.base,    //libevent的上下文
            listenCallback,                                   //接收到连接的回调函数
            m_args.base,                                      //回调函数获取的参数 arg
            LEV_OPT_REUSEABLE|LEV_OPT_CLOSE_ON_FREE,          //地址重用，evconnlistener关闭同时关闭socket
            10,                                               //连接队列大小，对应listen函数的backlog参数
            (sockaddr*)&sin,                                  //绑定的地址和端口
            sizeof(sin)
            );

        /* 创建IO线程 */
        if(pthread_create(&m_args.threadID, NULL, PollThread, &m_args) != 0)
        {
            cout << __FILE__<<":"<<__LINE__<<" pthread_create failed!" << endl;
            m_args.isRunning = false;
        }
        cout <<"Start http server at "<<m_args.ip<<":"<<m_args.port<<" ..."<<endl;

         m_args.isRunning = true;
    }

    return ret;
}

XTcpServer* XTcpServer::NewInstance(int threadCount,std::string ip, int port)
{
    XTcpServer * ret = new XTcpServer(threadCount, ip, port);
    if((ret == NULL)||!ret->construct())
    {
        delete ret;
        ret = NULL;
    }
    return ret;
}

void XTcpServer::listenCallback(struct evconnlistener * e, evutil_socket_t s, struct sockaddr *a, int socklen, void *arg)
{
    cout << "listen_cb" << endl;
    XTask *task = new XTcpTask(s);

    XThreadPool::GetInstance()->Dispatch(task);
}

void* XTcpServer::PollThread(void *arg)
{
    XTcpServerArgs * pArgs = (XTcpServerArgs*)arg;
    if(pArgs->base)
    {
        event_base_dispatch(pArgs->base);
    }

    if(pArgs->evConnListener)
    {
        evconnlistener_free(pArgs->evConnListener);
    }

    if(pArgs->base)
    {
        event_base_free(pArgs->base);
    }

    cout<<"exit form io thread"<<endl;
    return NULL;
}

XTcpServer::~XTcpServer()
{
    void *thrdRet = NULL;

    if(m_args.isRunning)
    {
        event_base_loopbreak(m_args.base);
        pthread_join(m_args.threadID, &thrdRet);
        m_args.isRunning = false;
    }
}
